package top.jfunc.websocket.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.exceptions.JedisConnectionException;
import top.jfunc.common.utils.ThreadFactoryBuilder;
import top.jfunc.websocket.TodoAtRemoved;
import top.jfunc.websocket.WebSocketManagerUtil;
import top.jfunc.websocket.memory.MemWebSocketManager;
import top.jfunc.websocket.redis.RedisWebSocketManager;
import top.jfunc.websocket.redis.Subscriber;

import java.util.concurrent.*;

/**
 * @author xiongshiyan at 2019/4/20 , contact me with email yanshixiong@126.com or phone 15208384257
 */
public class WebSocketManagerConfig {
    private static final Logger logger = LoggerFactory.getLogger(WebSocketManagerConfig.class);
    private static final Executor SUBSCRIBE_EXECUTOR;
    /**
     * 因为subscribe也会占用Redis的连接，所以不宜过大
     */
    private static final int MAX_RETRY_COUNT = 2;

    static {
        ThreadFactory threadFactory = ThreadFactoryBuilder.create()
                .setNameFormat("Redis-Subscribe-Thread-%d")
                .setDaemon(true)
                //.setThreadFactory(Thread::new)
                .build();
        //这个线程池的特点是线程会一直阻塞，直到有异常，
        //1. 所以core size=1，最大要设置为Integer.MAX_VALUE，SynchronousQueue阻塞队列一般要求maximumPoolSizes为无界(Integer.MAX_VALUE)，避免线程拒绝执行操作。
        //2. 线程空闲时间设置为60
        //3. 阻塞队列SynchronousQueue，不需要放在阻塞队列
        //4. 拒绝策略为抛异常
        //5. 建议不要监控太多channel，而是通过message来区分
        ThreadPoolExecutor.AbortPolicy abortPolicy = new ThreadPoolExecutor.AbortPolicy();
        SUBSCRIBE_EXECUTOR = new ThreadPoolExecutor(1,
                Integer.MAX_VALUE,
                60,
                TimeUnit.SECONDS,
                new SynchronousQueue<>(),
                threadFactory,
                abortPolicy);
    }
    /**
     * 基于内存的单机 WebSocketManager
     */
    public static void configMemory(){
        WebSocketManagerUtil.setWebSocketManager(new MemWebSocketManager());
    }
    /**
     * 基于Redis的集群 WebSocketManager
     */
    public static void configCluster(Jedis jedis){
        WebSocketManagerUtil.setWebSocketManager(new RedisWebSocketManager(jedis));
        SUBSCRIBE_EXECUTOR.execute(()->{
            for (int i = 0; i < MAX_RETRY_COUNT; i++) {
                try {
                    logger.info("redis subscribe {} index-{}", RedisWebSocketManager.CHANNEL, i);
                    //程序会阻塞在这里，理论上只会运行此代码一次，如果出现异常等相当于进行重试
                    jedis.subscribe(new Subscriber() , RedisWebSocketManager.CHANNEL);
                } catch (JedisConnectionException e) {
                    logger.info("redis subscribe {} error index-{}", RedisWebSocketManager.CHANNEL, i);
                    //如果发生了连接异常，则catch，然后重试
                    logger.error(e.getMessage(), e);
                }
            }

            logger.warn("redis subscribe had retried {}, won't accept more published messages", MAX_RETRY_COUNT);
        });
    }
    /*public static void configCluster(Jedis jedis){
        jedis.subscribe(new Subscriber() , RedisWebSocketManager.CHANNEL);
        WebSocketManagerUtil.setWebSocketManager(new RedisWebSocketManager(jedis));
    }*/

    /**
     * 配置心跳监测
     * @param period 多长时间执行一次
     * @param timeSpan 时间间隔
     * @param errorTolerant 错误容忍次数
     * @param todoAtRemoved 干什么
     */
    public static void configHeartBeatCheck(int period , long timeSpan , int errorTolerant , TodoAtRemoved todoAtRemoved){
        Executors.newScheduledThreadPool(1).scheduleAtFixedRate(()->{
            WebSocketHeartBeatChecker.INSTANCE.check(
                    WebSocketManagerUtil.getWebSocketManager() ,
                    timeSpan , errorTolerant, todoAtRemoved);
        } , 0 , period , TimeUnit.SECONDS);
    }
}
